var EventEmitter=require("events").EventEmitter,
    util=require("util"),
    Settings=require('./lib/mysettings.js'),
    kafka=require('kafka-node'),
    Producer=kafka.Producer,
    Client=kafka.Client,
    p= 0,a= 0;
    Consumer = kafka.Consumer;
    Snmper = require("./snmptest.js");

var Exe = function (){
    //EventEmitter.call(this);

    var config=new Settings();
    this.KAFKA_URL="";
    this.T_DEPLOY="";
    this.T_DEPLOY_RESULT="";
    this.kafka_client=null;
    this.kafka_producer=null;
    this.kafka_consumer=null;
    this.KAFKA_URL=config.KAFKA_URL;
    this.T_DEPLOY=config.T_DEPLOY;
    this.T_DEPLOY_RESULT=config.T_DEPLOY_RESULT;
    this.kafka_client=new Client(config.KAFKA_URL);
    this.kafka_producer=new Producer(this.kafka_client,{requireAcks:1});
    this.kafka_consumer = new Consumer(this.kafka_client, [{topic:this.T_DEPLOY}],{ autoCommit: true, fetchMaxWaitMs: 1000, fetchMaxBytes: 1024*1024 });
    this.snmper = new Snmper()
}

util.inherits(Exe,EventEmitter);

Exe.prototype.start = function(){
    var self = this

    self.on("result", function(data) {
        console.log(data);
        self.sendResult(null, data);
    })

    self.kafka_consumer.on('message', function (message) {
        var hosts = [], networks = [], str;
        console.log(message.offset);
        console.log(message.value);
        str = message.value;
        //console.log(str);
        hosts = JSON.parse(str);
        console.log(hosts);
        for ( var i in hosts){
            if (hosts[i].type == 'snmp'){
                networks.push(hosts[i])
            }
            //console.log(hosts[i]);
        }
        console.log(networks);
        self.snmper.doSnmp(self, networks)
        //console.log(hosts[0]);


    });

    this.kafka_producer.on('error',function(err){
        console.log('producer_error',err);
    });

    this.kafka_consumer.on('error', function (err) {
        console.log('consumer_error', err);
    });

}

Exe.prototype.sendResult = function(err,result) {
    //console.log(result);
    resultStr = JSON.stringify(result)

    this.kafka_producer.send(
        [{topic:this.T_DEPLOY_RESULT,partition:p,messages:resultStr,attributes:a}]
        ,function(err,res){
            if(err){
                console.log(err);
            }else{
                console.log(res);
            }
        })
}


function start() {

    var exe = new Exe();
    exe.start();
}

start();
